package com.atguigu.realtime.util

import com.atguigu.cm.constan.KafkaCons
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{KafkaUtils, _}

/**
 * 工具类类 MyKafkaUtil
 * author 剧情再美终是戏
 * mail 13286520398@163.com
 * date 2020/1/16 10:28
 * version 1.0
 **/
object MyKafkaUtil {

  // 定义kafka相关参数
  val kafkaParams = Map[String, Object](
    "bootstrap.servers" -> KafkaCons.BOOTSTRAP_SERVERS_CONFIG, // 连接kafka集群地址
    "key.deserializer" -> classOf[StringDeserializer],
    "value.deserializer" -> classOf[StringDeserializer],
    "group.id" -> KafkaCons.GROUP_SPARK_STREAM_PJ, // 消费者组
    "auto.offset.reset" -> "latest", // 从最后的位置开始消费
    "enable.auto.commit" -> (true: java.lang.Boolean) // 自动提交
  )


  // 获取kafka数据源
  def getDStream(ssc: StreamingContext, topics: Array[String]): InputDStream[ConsumerRecord[String, String]] = {
    KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent, // 标配. 只要 kafka 和 spark 没有部署在一台设备就应该是这个参数
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))
  }


}
